package com.gin.monitor.agg.distribution;


import com.gin.monitor.agg.utils.HashUtils;
import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.TrafficInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64NavigableMap;

import java.time.Duration;


/**
 * 违法车辆分析
 *
1620637885000,0002,75312,京L22188,190.1,27,31
1620637886000,0002,75312,京L22188,190.1,27,31
1620637889000,0002,75312,京L22188,190.1,27,31
1620637999000,0002,75312,京L22189,190.1,27,31
 *
 * @author gin
 * @date 2021/4/30
 */
public class AreaDistributionAnalysis2 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //第一种，当前的违法车辆（在5分钟内）如果已经出警了。（最后输出道主流中做删除处理）。
        // nc -lk 8888
        SingleOutputStreamOperator<TrafficInfo> stream = env.socketTextStream("node01", 8888)
                .map(new RichMapFunction<String, TrafficInfo>() {
                    @Override
                    public TrafficInfo map(String line) {
                        String[] s = line.split(",");
                        return new TrafficInfo(Long.parseLong(s[0]), s[1], s[2], s[3], Double.parseDouble(s[4]), s[5], s[6]);
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficInfo>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                        .withTimestampAssigner((event, timestamp) -> event.actionTime));


        stream.keyBy((KeySelector<TrafficInfo, String>) trafficInfo -> trafficInfo.areaId)
                //数据量大，Set集合去重会有内存溢出，采用bitmap/布隆过滤器(针对String类型, 取其hash code值)
                //默认情况：窗口没有触发，窗口中的所有数据都缓冲在状态（本地内存），也可能内存的溢出,自定义窗口的触发机制
        .timeWindow(Time.seconds(10))
                // 自定义 trigger 取代 window 默认实现
                // 自定义窗口触发器机制,默认窗口触发机制：窗口结束的时候触发，触发之后情况状态中的数据
                // 自定义：只要有一条数据进入窗口就触发，并且不保存数据到状态中
        .trigger(new Trigger<TrafficInfo, TimeWindow>() {
            @Override
            public TriggerResult onElement(TrafficInfo element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                // 当前窗口进入一条数据的回调函数
                // FIRE - 触发
                // PURGE - 清空状态
                return TriggerResult.FIRE_AND_PURGE;
            }

            @Override
            public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                // 当前窗口已经结束的回调函数(基于运行时间)
                // 因每来条就触发, 所以直接放行即可, 无需累计
                return TriggerResult.CONTINUE;
            }

            @Override
            public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                // 当前窗口已经结束的回调函数(基于事件时间)
                // 因每来条就触发, 所以直接放行即可, 无需累计
                return TriggerResult.CONTINUE;
            }

            @Override
            public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
                // 当前窗口对象销毁
                // 在 onElement 已经 PURGE 了, 不需要再销毁
            }
        })
        .process(new ProcessWindowFunction<TrafficInfo, String, String, TimeWindow>() {

            Roaring64NavigableMap bitMap = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                bitMap = new Roaring64NavigableMap();
            }

            @Override
            public void process(String k, Context context, Iterable<TrafficInfo> input, Collector<String> out) throws Exception {
                //一条数据执行一次，设计：在一个窗口中，每个区域对应一个布隆过滤器 ,为了方便位图计算，把布隆过滤器存入Redis中
                //设计一个Map集合，存放一个窗口中，每个区域的车辆(去重之后)数量
                long start = context.window().getStart();
                long end = context.window().getEnd();

                for (TrafficInfo trafficInfo : input) {
                    bitMap.add(HashUtils.fnvHash(trafficInfo.car));
                }

                out.collect("区域" + k + ",在窗口期开始时间" + start
                        + ",到窗口结束时间" + end + " ,一共有: " + bitMap.getIntCardinality() + " 辆车");
            }
        })
        .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
